package com.peng.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * KafkaProducer
 *
 * @author: lupeng6
 * @create: 2023/3/18 13:54
 */
public class MyKafkaConsumer {

    // topic名字
    private static final String TOPIC_NAME = "sample2";
    // 消费者组名字
    private static final String CONSUMER_GROUP = "testGroup1";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        // broker列表
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.14:9092,192.168.1.14:9093,192.168.1.14:9094");
        // consumerGroup
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
        // key反序列化方式
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // value反序列化方式
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // consumer 一次拉取多少条数据，默认500，如果consumer的等待时间内没有拿到500则继续poll
        // 如果拿到了500条数据则执行消费者的逻辑
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        // 如果两次poll的时间大于30秒则kafka会任务消费者的消费能力差，将consumer踢出消费者组，将partition给其他消费者
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        // consumer给broker发送心跳的时间间隔
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        // kafka如果超过10秒没有收到消费者的心跳，则会吧消费者提出组，进行rebalance，将分区交给其他的消费者
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
        // kafka消费者分区分配策略
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StickyAssignor.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 消费那个topic
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("收到消息：partition=%s offset=%s key=%s value=%s%n",
                        record.partition(), record.offset(), record.key(), record.value());
            }
            // 想kafka-broker提交poll消息时候回去到的offset
            consumer.commitSync();
            if (records.count() > 0) {
                // 手动异步提交offset，当前线程提交offset不会阻塞
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + Arrays.toString(exception.getStackTrace()));
                        }
                    }
                });
            }
        }
    }
}
